package workflow

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/asynkron/protoactor-go/actor"
	"github.com/louloulin/dataflare/pkg/actor/system"
	"github.com/louloulin/dataflare/pkg/workflow/connector"
)

// SinkActor is an actor that writes data to a sink
type SinkActor struct {
	// id is the sink ID
	id string
	// definition is the sink definition
	definition SinkDefinition
	// instanceID is the workflow instance ID
	instanceID string
	// stateManager is the state manager PID
	stateManager *actor.PID
	// metricsCollector is the metrics collector PID
	metricsCollector *actor.PID
	// connector is the sink connector
	connector connector.SinkConnector
	// status is the node status
	status string
	// mu is a mutex for protecting the status
	mu sync.RWMutex
	// logger is the sink actor logger
	logger *log.Logger
	// running indicates if the sink is running
	running bool
	// writtenCount is the number of records written
	writtenCount int64
	// errorCount is the number of errors
	errorCount int64
	// lastError is the last error
	lastError error
	// lastWriteTime is the last time data was written
	lastWriteTime time.Time
	// sourceActors are the source actors that send data to this sink
	sourceActors map[string]*actor.PID
}

// NewSinkActor creates a new sink actor
func NewSinkActor(id string, definition SinkDefinition, instanceID string, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	return &SinkActor{
		id:               id,
		definition:       definition,
		instanceID:       instanceID,
		stateManager:     stateManager,
		metricsCollector: metricsCollector,
		sourceActors:     make(map[string]*actor.PID),
		status:           "created",
		logger:           log.New(log.Writer(), fmt.Sprintf("[SinkActor-%s-%s] ", instanceID, id), log.LstdFlags),
	}
}

// Receive handles messages for the sink actor
func (s *SinkActor) Receive(context actor.Context) {
	switch msg := context.Message().(type) {
	case *actor.Started:
		s.handleStarted(context)
	case *actor.Stopping:
		s.handleStopping(context)
	case *StartNodeCommand:
		s.handleStartNode(context, msg)
	case *StopNodeCommand:
		s.handleStopNode(context)
	case *GetNodeStatusCommand:
		s.handleGetNodeStatus(context)
	case *system.DataMessage:
		s.handleDataMessage(context, msg)
	default:
		s.logger.Printf("Received unknown message: %T", msg)
	}
}

// handleStarted handles the Started message
func (s *SinkActor) handleStarted(context actor.Context) {
	s.logger.Println("Sink actor started")

	// Create connector
	var err error
	s.connector, err = s.createConnector()
	if err != nil {
		s.logger.Printf("Failed to create connector: %v", err)
		s.status = "failed"
		context.Send(context.Parent(), &NodeErrorMessage{
			NodeID: s.id,
			Error:  fmt.Sprintf("Failed to create connector: %v", err),
		})
		return
	}

	// Report metrics
	if s.metricsCollector != nil {
		context.Send(s.metricsCollector, &system.ReportMetricCommand{
			Name:  "sink.started",
			Value: 1,
			Tags: map[string]string{
				"sink_id":     s.id,
				"instance_id": s.instanceID,
				"type":        s.definition.Type,
			},
		})
	}

	s.status = "initialized"
	s.logger.Println("Sink actor initialized")
}

// handleStopping handles the Stopping message
func (s *SinkActor) handleStopping(context actor.Context) {
	s.logger.Println("Sink actor stopping")

	// Stop connector
	if s.connector != nil {
		err := s.connector.Stop(context.Context())
		if err != nil {
			s.logger.Printf("Failed to stop connector: %v", err)
		}
	}

	// Report metrics
	if s.metricsCollector != nil {
		context.Send(s.metricsCollector, &system.ReportMetricCommand{
			Name:  "sink.stopped",
			Value: 1,
			Tags: map[string]string{
				"sink_id":     s.id,
				"instance_id": s.instanceID,
				"type":        s.definition.Type,
			},
		})
	}

	s.status = "stopped"
}

// handleStartNode handles the StartNodeCommand message
func (s *SinkActor) handleStartNode(context actor.Context, msg *StartNodeCommand) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Check if already running
	if s.running {
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   "sink already running",
		})
		return
	}

	// Initialize connector
	err := s.connector.Init(context.Context(), s.definition.Config)
	if err != nil {
		s.logger.Printf("Failed to initialize connector: %v", err)
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to initialize connector: %v", err),
		})
		return
	}

	// Start connector
	err = s.connector.Start(context.Context())
	if err != nil {
		s.logger.Printf("Failed to start connector: %v", err)
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to start connector: %v", err),
		})
		return
	}

	// Update state
	s.running = true
	s.status = "running"

	// Respond with success
	context.Respond(&StartNodeResponse{
		Success: true,
	})

	// Report status change
	context.Send(context.Parent(), &NodeStatusChangedMessage{
		NodeID: s.id,
		Status: "running",
	})

	s.logger.Println("Sink started")
}

// handleStopNode handles the StopNodeCommand message
func (s *SinkActor) handleStopNode(context actor.Context) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Check if running
	if !s.running {
		context.Respond(&StopNodeResponse{
			Success: false,
			Error:   "sink not running",
		})
		return
	}

	// Stop connector
	err := s.connector.Stop(context.Context())
	if err != nil {
		s.logger.Printf("Failed to stop connector: %v", err)
		context.Respond(&StopNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to stop connector: %v", err),
		})
		return
	}

	// Update state
	s.running = false
	s.status = "stopped"

	// Respond with success
	context.Respond(&StopNodeResponse{
		Success: true,
	})

	// Report status change
	context.Send(context.Parent(), &NodeStatusChangedMessage{
		NodeID: s.id,
		Status: "stopped",
	})

	s.logger.Println("Sink stopped")
}

// handleGetNodeStatus handles the GetNodeStatusCommand message
func (s *SinkActor) handleGetNodeStatus(context actor.Context) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// Create metrics
	metrics := map[string]interface{}{
		"written_count":   s.writtenCount,
		"error_count":     s.errorCount,
		"last_write_time": s.lastWriteTime,
	}

	// Respond with status
	context.Respond(&GetNodeStatusResponse{
		Status:  s.status,
		Metrics: metrics,
		Error:   s.lastError,
	})
}

// handleDataMessage handles the DataMessage message
func (s *SinkActor) handleDataMessage(context actor.Context, msg *system.DataMessage) {
	s.mu.RLock()
	running := s.running
	s.mu.RUnlock()

	// Check if running
	if !running {
		s.logger.Println("Received data message while not running, ignoring")
		
		// Return credit to source
		if sourceID, ok := msg.Metadata["source_id"]; ok {
			if sourcePID, ok := s.sourceActors[sourceID]; ok {
				context.Send(sourcePID, &CreditReturnedMessage{})
			}
		}
		
		return
	}

	// Create record
	record := connector.Record{
		Key:   []byte(msg.Metadata["record_key"]),
		Value: msg.Data,
	}

	// Write data
	startTime := time.Now()
	err := s.connector.Write(context.Context(), record)
	writingTime := time.Since(startTime)

	// Update state
	s.mu.Lock()
	s.lastWriteTime = time.Now()
	s.mu.Unlock()

	// Handle error
	if err != nil {
		s.logger.Printf("Failed to write data: %v", err)
		s.mu.Lock()
		s.lastError = err
		s.errorCount++
		s.mu.Unlock()

		// Report error metrics
		if s.metricsCollector != nil {
			context.Send(s.metricsCollector, &system.ReportMetricCommand{
				Name:  "sink.errors",
				Value: 1,
				Tags: map[string]string{
					"sink_id":     s.id,
					"instance_id": s.instanceID,
					"type":        s.definition.Type,
				},
			})
		}
	} else {
		// Update written count
		s.mu.Lock()
		s.writtenCount++
		s.mu.Unlock()

		// Report metrics
		if s.metricsCollector != nil {
			context.Send(s.metricsCollector, &system.ReportMetricCommand{
				Name:  "sink.records_written",
				Value: 1,
				Tags: map[string]string{
					"sink_id":     s.id,
					"instance_id": s.instanceID,
					"type":        s.definition.Type,
				},
			})

			context.Send(s.metricsCollector, &system.ReportMetricCommand{
				Name:  "sink.writing_time",
				Value: float64(writingTime.Milliseconds()),
				Tags: map[string]string{
					"sink_id":     s.id,
					"instance_id": s.instanceID,
					"type":        s.definition.Type,
				},
			})
		}
	}

	// Return credit to source
	if sourceID, ok := msg.Metadata["source_id"]; ok {
		if sourcePID, ok := s.sourceActors[sourceID]; ok {
			context.Send(sourcePID, &CreditReturnedMessage{})
		} else if processorID, ok := msg.Metadata["processor_id"]; ok {
			// If we don't have the source actor, but we have a processor ID,
			// we can try to return credit to the processor
			if processorPID, ok := s.sourceActors[processorID]; ok {
				context.Send(processorPID, &CreditReturnedMessage{})
			}
		}
	}
}

// createConnector creates a sink connector
func (s *SinkActor) createConnector() (connector.SinkConnector, error) {
	// Get connector factory
	factory := connector.GetSinkConnectorFactory(s.definition.Type)
	if factory == nil {
		return nil, fmt.Errorf("unknown sink connector type: %s", s.definition.Type)
	}

	// Create connector
	return factory()
}
